package com.tuniondata.netty.client.future;

import com.tuniondata.netty.bean.RpcMessage;
import com.tuniondata.netty.common.protocol.Protocol;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class SyncRequest {
    private final Logger logger = LoggerFactory.getLogger(SyncRequest.class);

    public RpcMessage writeAndSync(final Channel channel,
                                   final RpcMessage request,
                                   final long timeout) throws Exception {

        if (channel == null) {
            throw new NullPointerException("channel");
        }
        if (request == null) {
            throw new NullPointerException("request");
        }
        if (timeout <= 0) {
            throw new IllegalArgumentException("timeout <= 0");
        }

        String requestId = UUID.randomUUID().toString();
        request.setRequestId(requestId);

        RequestFuture<RpcMessage> future = new SyncRequestFuture(request.getRequestId());

        SyncRequestMap.syncKey.put(request.getRequestId(), future);

        logger.info("writeAndSync request, requestId is {}", request.getRequestId());

        RpcMessage response = doWriteAndSync(channel, request, timeout, future);

        logger.info("writeAndSync response, requestId is {}", response.getRequestId());

        SyncRequestMap.syncKey.remove(request.getRequestId());

        return response;
    }

    private RpcMessage doWriteAndSync(final Channel channel,
                                    final RpcMessage request,
                                    final long timeout,
                                    final RequestFuture<RpcMessage> writeFuture) throws Exception {
        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            public void operationComplete(ChannelFuture future) throws Exception {
                logger.info("doWriteAndSync response, requestId is {}", writeFuture.requestId());
                writeFuture.setWriteResult(future.isSuccess());
                writeFuture.setCause(future.cause());
                //失败移除
                if (!writeFuture.isWriteSuccess()) {
                    SyncRequestMap.syncKey.remove(writeFuture.requestId());
                }
            }
        });

        RpcMessage response = writeFuture.get(timeout, TimeUnit.MILLISECONDS);
        if (response == null) {
            if (writeFuture.isTimeout()) {
                throw new TimeoutException();
            } else {
                throw new Exception(writeFuture.cause());
            }
        }

        return response;
    }
}
